package com.wffz.config.redis;


import com.alibaba.fastjson.JSONObject;
import com.wffz.po.Bet;
import com.wffz.service.PatchService;
import com.wffz.vo.PrizesVO;
import com.wffz.websocket.BetSocket;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.websocket.EncodeException;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author: kqyin
 * @date: 2023/1/9 15:14
 * @Description: redis延迟队列配置
 */
@Configuration
// 当redission.delayqueue.enable属性配置为true时，该配置才会生效，也就是启用redission的延迟队列
@ConditionalOnProperty(value = "redission.delayqueue.enable")
@Slf4j
public class RedisDelayQueueConfig {


    @Autowired
    private PatchService patchService;

    /**
     * @Author kqyin
     * @Date 2023/1/9 19:15
     * @Description 线程池
     */
    @Bean("delayExecutor")
    public ExecutorService getDelayExecutor() {
        return Executors.newFixedThreadPool(5);
    }

    /**
     * @Author kqyin
     * @Date 2023/1/9 19:15
     * @Description 获取延迟队列组，并执行
     */
    @Bean
    public RedisDelayQueueEnum[] startRedisDelayQueue(ApplicationContext applicationContext, RedissonClient redissonClient,
                                                      @Qualifier("delayExecutor") ExecutorService executorService) {
        RedisDelayQueueEnum[] redisDelayQueueArr = RedisDelayQueueEnum.values();
        for (RedisDelayQueueEnum queueEnum : redisDelayQueueArr) {
            startThread(applicationContext, redissonClient, executorService, queueEnum);
        }
        return redisDelayQueueArr;
    }

    /**
     * @Author kqyin
     * @Date 2023/1/9 19:15
     * @Description 执行延迟队列
     */
    private <T> void startThread(ApplicationContext applicationContext, RedissonClient redissonClient,
                                 ExecutorService executorService, RedisDelayQueueEnum queueEnum) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueEnum.getQueueName());
        // 由于此线程需要常驻，可以新建线程，不用交给线程池管理
        Thread thread = new Thread(() -> {
            log.info("启动监听延迟队列线程:{}", queueEnum.getQueueName());
            CopyOnWriteArraySet<BetSocket> webSocketSet = BetSocket.getWebSocketSet();
            while (true) {
                try {
                    // 获取到执行类
                    RedisDelayQueueHandler redisDelayQueueHandler = (RedisDelayQueueHandler) applicationContext.getBean(queueEnum.getBeanId());
                    // 获取到被执行对象，作为参数传递给redisDelayQueueHandler
                    T t = blockingFairQueue.take();
                    log.info("监听延迟队列线程:{}, 获取到值:{}", queueEnum.getQueueName(), JSONObject.toJSONString(t));
                    webSocketSet.forEach(c -> {
                        try {
                            // 如果订单是当前登录用户的，发送数据到前端
                            Bet b = (Bet) t;
                            String sessionUserName = c.getSession().getUserPrincipal().getName();
                            if (b.getLoginName().equals(sessionUserName)) {
                                // 获取开奖信息
                                // 中奖金额
                                PrizesVO vo = patchService.PrizesResult(b);
                                log.info("开奖信息：" + vo.getPrizesResult());
                                String prizes = String.format("%.2f", vo.getPrizesResult());
                                vo.setPrizes(prizes);
                                try {
                                    c.sendObjMessage(vo);
                                } catch (EncodeException e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    });
                    executorService.submit(() -> redisDelayQueueHandler.exec(t));
                } catch (Exception e) {
                    log.error("监听延迟队列线程错误,", e);
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException ex) {
                    }
                }
            }
        });
        thread.setName(queueEnum.getQueueName());
        thread.start();
    }
}
